package net.bwie.jtp.dim.job;

import net.bwie.jtp.dim.function.DimDataFilterFunction;
import net.bwie.jtp.dim.function.DimHbaseSinkFunction;
import net.bwie.jtp.dim.function.HbaseDimMapFunction;
import net.bwie.realtime.guanjuntao.util.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

public class RealtimeSyncDataDimmJob {
    public static void main(String[] args) throws Exception {

        // TODO: 2025/6/2 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing( 5000L);
        // TODO: 2025/6/2 拉取数据 
        DataStream<String> stringDataStream = KafkaUtil.consumerKafka(env, "topic-db");
        // TODO: 2025/6/2  数据转换
        DataStream<String> dimStreame = handle(stringDataStream);
        // TODO: 2025/6/2 接收数据
        dimStreame.addSink(new DimHbaseSinkFunction());
        // TODO: 2025/6/2 触发执行环境

        env.execute("RealtimeSyncDataDimmJob");

    }
    /*
       按照表名称过滤获取对应维表数据流,保存Hbase,列簇
     */
    private static DataStream<String> handle(DataStream<String> stringDataStream) {
        HashMap<String, String> dimMap = new HashMap<String,String>(){
            {
                put("base_dic","dic_code");put("base_category1","id");put("base_category2","id");
                put("base_category3","id");put("base_province","id");put("base_region","id");
                put("base_trademark","id");put("activity_info","id");put("activity_rule","id");
                put("coupon_info","id");put("sku_info","id");put("spu_info","id");put("user_info","id");

            }
        };

        // TODO: 2025/6/2 维度名称
        Set<String> dimSet = new HashSet<>(dimMap.keySet());

        // TODO: 2025/6/2 过滤维度表数据
        SingleOutputStreamOperator<String> filter = stringDataStream.filter(new DimDataFilterFunction(dimSet));
        // TODO: 2025/6/2 添加字段值
        SingleOutputStreamOperator<String> map = filter.map(new HbaseDimMapFunction(dimMap));

        return map;
    }






























}
